package com.alison.sparkstream.source

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies};

object E4_stream_kafka {

  def main() = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

    // 初始化 StreamingContext，设置微批处理的时间间隔为 3 秒
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    val topic = "spark-topic" // Kafka主题
    val brokers = "192.168.56.104:9092" // Kafka集群地址

    //定义 Kafka 参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
      ConsumerConfig.GROUP_ID_CONFIG -> topic,
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    // 读取 Kafka 数据创建 DStream
    val stream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream[String, String](ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Set(topic), kafkaPara))

    // Word Count统计
    val wordCounts = stream
      .map(record => record.value()) // 将每条消息的 value 取出（不需要key）
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)

    // 打印Word Count结果
    wordCounts.print()

    // 开启任务
    ssc.start()

    // 等待应用程序终止
    ssc.awaitTermination()
  }
}
